import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;


public class ClusterNode extends Node
{
/*---------------- MESSAGE CONSTANTS -----------------*/
    /** keep alive, sent from parent to child node */
    public static final String KEEP_ALIVE = "KeepAlive";

/*---------------- STATIC MEMBERS -----------------*/
    /** Prefix for control messages */
    public static final String CTL_PRE = "c_";
    
    /** Prefix for peer deletion */
    public static final String CTL_DEL = "d_";
    
    /** Prefix for peer deletion stop */
    public static final String CTL_DEL_STOP = "ds_";
    
    /** Prefix for adding yourself to child list */
    public static final String CTL_ADDME = "am_";
    
    /** Prefix for joining */
    public static final String CTL_JOIN = "j_";
    
    /** Prefix for new peer */
    public static final String CTL_PEER = "p_";
    
    /** Prefix for checking aliveness of children */
    public static final String CTL_ALIVE = "a_";
    
    /** Prefix for promotion messages */
    public static final String CTL_PROMOTE = "pr_";
    
    /** Prefix for responding to alive */
    public static final String CTL_ACK = "ack_";
    
    /** Prefix for parent peer message passing */
    public static final String CTL_PARENTPEER = "pp_";
    
    /** Prefix for changing parent */
    public static final String CTL_CHANGEPARENT = "cp_";
    
    /** index of message integer in a message.split() */
    public static final int MSG_IDX = 0;
    
    /** index of message generator in a message.split() */
    public static final int ORG_IDX = 1;
    
    /** index of message sender in a message.split() */
    public static final int SND_IDX = 2;
    
    /** maximum number of ms to wait for a keep alive
    * before finding a new parent */
    public static final long KA_MAX = 6000;
/*---------------- INSTANCE MEMBERS -----------------*/
    /** All peers of our parent aka other super nodes etc... */
    public Set<Integer> parentPeers;
    
    /** Current parent of ourselves */
    public ClusterNode parent;
    
    /** All children id numbers */
    public Set<Integer> childrenId;
    
    /** All of your peer nodes connected to your shared parent */
    public Set<Integer> peerId;
    
    /** mesages since the last confirmation from the parent */
    public Set<String> savedMessages;
    
    public int numControlMessages;
    
    public long lastKeepAlive;
    
    public long promoteKeepAlive = -1;
    
    public long promoteWait = 1;
    
    public Integer toPromote = -1;
/*---------------- CONSTRUCTORS -----------------*/
    public ClusterNode(int ID)
    {
        super(ID);
        parent = null;
        parentPeers = new HashSet<Integer>();
        childrenId = new HashSet<Integer>();
        peerId = new HashSet<Integer>();
        savedMessages = new HashSet<String>();
        numControlMessages = 0;
        
        // assume just found parent
        lastKeepAlive = System.currentTimeMillis();
    }
    
    public ClusterNode(int ID, Node connect)
    {
        super(ID);
        parent = (ClusterNode)connect;
        if(parent != null)
            fullConnect(parent);
        parentPeers = new HashSet<Integer>();
        childrenId = new HashSet<Integer>();
        peerId = new HashSet<Integer>();
        savedMessages = new HashSet<String>();
        numControlMessages = 0;
        setPeers(parent.childrenId);
        ((ClusterNode)connect).childrenId.add((Integer)ID);
        setParentPeers(((ClusterNode)connect).peerId);
        
        // assume just found parent
        lastKeepAlive = System.currentTimeMillis();
    }
    
/*---------------- NEIGHBOR MUTATORS -----------------*/
    /** Clear and update the peers to the given list.
    */
    public void setPeers(Set<Integer> peers)
    {
        peerId.clear();
        for(Integer i : peers)
        {
            if(i != nodeId)
                peerId.add(i);
        }
    }
    
    /** clear and update the children to the given list
    */
    public void setChildren(Set<Integer> children)
    {
        childrenId.clear();
        for(Integer i : children)
            childrenId.add(i);
    }
    
    /** clear and update the parent peers to the given list.
    */
    public void setParentPeers(Set<Integer> parPeers)
    {
        parentPeers.clear();
        
        for(Integer n : parPeers)
        {
            if(n != parent.nodeId)
                parentPeers.add(n);
        }
    }
    
/*---------------- OVERWRITTEN PARENT METHODS -----------------*/
    @Override
    /** broadcasts a new meessage up and down the tree.
    * This method should only be called when a new message is generated.
    * forwarding messages depends on the sender of the message.
    */
    public void send(String message)
    {
        if(parent == null)
        {
            // do super node stuff
        }
        
        // do normal node stuff:
        
        // sign message
        message = message + "," + nodeId;
        
        // send up and down the tree
        sendUp(message);
        sendDown(message);
    }
    
    @Override
    public void processMessage(String message)
    {
        // process control messages
        if(message.startsWith(CTL_PRE))
        {
            // register it was control
            numControlMessages++;
            
            message = message.substring(CTL_PRE.length());
            
            // check if it is alive message
            if(message.startsWith(CTL_ALIVE))
            {
            	message = message.substring(CTL_ALIVE.length());
            	
            	commChannels.get(Integer.parseInt(message)).add(CTL_PRE+CTL_ACK+nodeId);
            }
            
            // check if it is a parent peer message
            if(message.startsWith(CTL_PARENTPEER))
            {
            	message = message.substring(CTL_PARENTPEER.length());
            	String [] newParentPeers = null;
            	if(message.contains(","))
            	{
            		newParentPeers = message.split(",");
            	}
            	else
            	{
            		newParentPeers = new String []{message};
            	}
            	for(String s : newParentPeers)
            	{
            		System.out.println("S: "+s);
            		if(!s.equals("") && (Integer.parseInt(s) != parent.nodeId))
            		{
            			parentPeers.add(Integer.parseInt(s));
            		}
            	}
            }
            
            // if message is add me
            if(message.startsWith(CTL_ADDME))
            {
            	message = message.substring(CTL_ADDME.length());
            	childrenId.add(Integer.parseInt(message));
            }
            
            // check if it a change parent message
            if(message.startsWith(CTL_CHANGEPARENT))
            {
            	message = message.substring(CTL_CHANGEPARENT.length());
            	parent = (ClusterNode) Node.globalNodes.get(Integer.parseInt(message));
            }
            
            // check if it is a promote message
            if(message.startsWith(CTL_PROMOTE))
            {
            	message = message.substring(CTL_PROMOTE.length());
            	String [] parts = message.split("\\.");
            	String [] firstHalf = parts[0].split(",");
            	String [] secondHalf = parts[1].split(",");
            	parent = (ClusterNode) Node.globalNodes.get(Integer.parseInt(firstHalf[0]));
            	System.out.println("Parent set to: "+parent.nodeId);
            	peerId.clear();
            	peerId.add(Integer.parseInt(secondHalf[secondHalf.length-1]));
            	for(int i = 1; i < firstHalf.length; i++)
            	{
            		childrenId.add(Integer.parseInt(firstHalf[i]));
            	}
            	for(Integer node : childrenId)
            	{
            		commChannels.get(node).add(CTL_PRE+CTL_CHANGEPARENT+nodeId);
            	}
            	
            	for(Integer node : childrenId)
            	{
            		commChannels.get(node).add(CTL_PRE+CTL_PARENTPEER+secondHalf[secondHalf.length-1]);
            	}
            	
            	parentPeers.clear();
            	for(int i = 0; i < secondHalf.length-1; i++)
            	{
            		parentPeers.add(Integer.parseInt(secondHalf[i]));
            	}
            	parent.childrenId.add(nodeId);
            	System.out.println(nodeId+" has been promoted");
            	System.out.println("Peers:");
            	for(Integer node : peerId)
            	{
            		System.out.println("\t"+node);
            	}
            	System.out.println("Children:");
            	for(Integer node : childrenId)
            	{
            		System.out.println("\t"+node);
            	}
            	toPromote = -1;
            	promoteKeepAlive = -1;
    			promoteWait = 1;
            }
            
            // check if it is an ack message
            if(message.startsWith(CTL_ACK))
            {
            	message = message.substring(CTL_ACK.length());
            	toPromote = Integer.parseInt(message);
        		Integer [] childList = new Integer [childrenId.size()];
        		childrenId.toArray(childList);
    			childrenId.remove(toPromote);
    			childrenId.toArray(childList);
    			Set<Integer> toMove = new HashSet<Integer>();
    			for(int i = 0; i < childList.length/2; i++)
    			{
    				Integer toRemove = childList[i];
    				childrenId.remove(toRemove);
    				toMove.add(toRemove);
    			}
    			peerId.add(toPromote);
    			String promoteMessage = CTL_PRE + CTL_PROMOTE + parent.nodeId + ",";
    			for(Integer node : toMove)
    			{
    				promoteMessage += node.toString();
    				promoteMessage += ",";
    			}
    			promoteMessage += ".";
    			String parentPeerMessage = "";
    			for(Integer node : parentPeers)
    			{
    				promoteMessage += node.toString();
    				promoteMessage += ",";
    				parentPeerMessage += node.toString();
    				parentPeerMessage += ",";
    			}
    			parentPeerMessage += toPromote;
    			promoteMessage = promoteMessage + nodeId;
    			if(parentPeerMessage.length() != 0)
    			{
    				parentPeerMessage = parentPeerMessage.substring(0,parentPeerMessage.length());
    			}
    			for(Integer node : childrenId)
    			{
    				System.out.println(parentPeerMessage);
    				commChannels.get(node).add(CTL_PRE+CTL_PARENTPEER+parentPeerMessage);
    			}
    			commChannels.get(toPromote).add(promoteMessage);
    			System.out.println("Node "+nodeId);
    			toPromote = -1;
    			promoteKeepAlive = -1;
    			promoteWait = 1;
            }
            
            if(message.startsWith(CTL_DEL_STOP))
            {
            	message = message.substring(CTL_DEL_STOP.length());
            	Integer deleteNode = Integer.parseInt(message);
            	if(childrenId.contains(deleteNode))
            	{
            		childrenId.remove(deleteNode);
            	}
            	
            	if(peerId.contains(deleteNode))
            	{
            		peerId.remove(deleteNode);
            	}
            	
            	if(parentPeers.contains(deleteNode))
            	{
            		parentPeers.remove(deleteNode);
            	}
            	
            	if(peerId.size() == 0 && promoteKeepAlive == -1)
            	{
            		if(System.currentTimeMillis() > promoteWait && promoteWait != 1)
            		{
	            		Integer [] childList = null;
	            		if(toPromote == -1 && childrenId.size() != 0)
	            		{
		            		Random rand = new Random();
		            		childList = new Integer [childrenId.size()];
		            		childrenId.toArray(childList);
		            		toPromote = childList[rand.nextInt(childList.length)];
		            		commChannels.get(toPromote).add(CTL_PRE+CTL_ALIVE+nodeId);
		            		promoteKeepAlive = System.currentTimeMillis();
	            		}
            		}
            		else if(promoteWait == 1)
            		{
            			promoteWait = System.currentTimeMillis() + 7000;
            		}
            	}
            }
            
            // check if this is delete message
            if(message.startsWith(CTL_DEL))
            {
            	message = message.substring(CTL_DEL.length());
            	Integer deleteNode = Integer.parseInt(message);
            	if(childrenId.contains(deleteNode))
            	{
            		childrenId.remove(deleteNode);
            	}
            	
            	if(peerId.contains(deleteNode))
            	{
            		peerId.remove(deleteNode);
            	}
            	
            	if(parentPeers.contains(deleteNode))
            	{
            		parentPeers.remove(deleteNode);
            	}
            	
            	if(peerId.size() == 0 && promoteKeepAlive == -1)
            	{
            		System.out.println("Del check "+(System.currentTimeMillis() > promoteWait && promoteWait != 1)+" "+(System.currentTimeMillis() - promoteWait));
            		if(System.currentTimeMillis() > promoteWait && promoteWait != 1)
            		{
	            		Integer [] childList = null;
	            		if(toPromote == -1 && childrenId.size() != 0)
	            		{
		            		Random rand = new Random();
		            		childList = new Integer [childrenId.size()];
		            		childrenId.toArray(childList);
		            		toPromote = childList[rand.nextInt(childList.length)];
		            		commChannels.get(toPromote).add(CTL_PRE+CTL_ALIVE+nodeId);
		            		promoteKeepAlive = System.currentTimeMillis();
	            		}
            		}
            		else if(promoteWait == 1)
            		{
            			System.out.println("Waiting to promote DEL");
            			promoteWait = System.currentTimeMillis();
            			try {
							Thread.sleep(4000);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
            		}
            	}
            	
            	for(Integer child : childrenId)
            	{
            		commChannels.get(child).add(CTL_PRE+CTL_DEL_STOP+deleteNode);
            	}
            	
            	for(Integer peer : peerId)
            	{
            		commChannels.get(peer).add(CTL_PRE+CTL_DEL_STOP+deleteNode);
            	}
            }
            
            // check if we received a join message
            if(message.startsWith(CTL_JOIN))
            {
            	message = message.substring(CTL_JOIN.length());
            	
            	Integer newNode = Integer.parseInt(message);
            	
            	for(Integer child : childrenId)
            	{
            		commChannels.get(child).add(CTL_PRE+CTL_PEER+newNode);
            	}
            	
            	childrenId.add(newNode);
            }
            
         // check if we received a peer message
            if(message.startsWith(CTL_PEER))
            {
            	message = message.substring(CTL_PEER.length());
            	
            	Integer newNode = Integer.parseInt(message);
            	peerId.add(newNode);
            	
            	for(String s : savedMessages)
            	{
            		sendUp(s);
            	}
            	savedMessages.clear();
            }
            
            // keep alive from parent received
            if(message.equals(KEEP_ALIVE))
            {
                lastKeepAlive = System.currentTimeMillis();
                savedMessages.clear();
                return;
            }
            
            if(parent == null)
            {
                // do super node stuff
            }
            
            // do normal node stuff
            
            return;
        }
        
        // process data message
        String[] split = message.split(",");
        
        // don't do anything with copied messages
        String base = split[MSG_IDX] + "," + split[ORG_IDX];
        if(information.contains(base))
        {
            copyCount++;
            return;
        }
        information.add(base);
        
        // got from parent, send to children
        if(parent != null && Integer.parseInt(split[SND_IDX]) == parent.nodeId)
        {
            // update sender
            split[SND_IDX] = "" + nodeId;
            String forward = mashSplit(split);
            // update keep alive
            lastKeepAlive = System.currentTimeMillis();
            savedMessages.clear();
            // forward
            sendDown(forward);
            return;
        }

        
        // message received from child node
        if(childrenId.contains(Integer.parseInt(split[SND_IDX])))
        {
            // change sender
            split[SND_IDX] = "" + nodeId;
            String forward = mashSplit(split);
            // forward
            sendUp(forward);
            sendDown(forward);
        }
        
        // got a regular message from someone else?
    }
    
    @Override
    public void step()
    {
        // send keep alive to overlay children
        sendDown(CTL_PRE + KEEP_ALIVE);
        
        // check if promote keep alive fails
        if(nodeId == 3)
        {
        	System.out.println(promoteKeepAlive);
        }
        /*if((promoteKeepAlive != -1 && System.currentTimeMillis() > promoteKeepAlive + KA_MAX) || promoteWait != 1)
        {
        	System.out.println("Step check "+(System.currentTimeMillis() > promoteWait && promoteWait != 1));
        	if(System.currentTimeMillis() > promoteWait && promoteWait != 1)
    		{
        		Integer [] childList = null;
        		if(toPromote == -1 && childrenId.size() != 0)
        		{
            		Random rand = new Random();
            		childList = new Integer [childrenId.size()];
            		childrenId.toArray(childList);
            		toPromote = childList[rand.nextInt(childList.length)];
            		commChannels.get(toPromote).add(CTL_PRE+CTL_ALIVE+nodeId);
            		promoteKeepAlive = System.currentTimeMillis();
        		}
    		}
    		else if(promoteWait == 1)
    		{
    			System.out.println("Waiting to promote step()");
    			promoteWait = System.currentTimeMillis() + 7000;
    		}
        }*/
        
        // check keep alive status of overlay parent
        if(parent != null && System.currentTimeMillis() > lastKeepAlive + KA_MAX)
        {
        	System.out.println("Parent Peers: "+parentPeers.toString());
        	Random rand = new Random();
        	if(parentPeers.size() == 0)
        	{
        		System.out.println("No more parents for node "+nodeId);
        		System.out.println(parent.nodeId);
        		return;
        	}
        	Integer [] parentPeerList = new Integer [parentPeers.size()];
            parentPeers.toArray(parentPeerList);
            Integer newParent = parentPeerList[rand.nextInt(parentPeerList.length)];
            parentPeers.remove(newParent);
            LinkedBlockingQueue<String> tmpChannel = commChannels.get(newParent);
            tmpChannel.add(CTL_PRE+CTL_DEL+parent.nodeId);
            tmpChannel.add(CTL_PRE+CTL_JOIN+nodeId);
            parent = (ClusterNode) Node.globalNodes.get(newParent);
            lastKeepAlive = System.currentTimeMillis();
        }
        
        // do anything needed for supernode status
    }
    
/*---------------- PROCESSMESSAGE HELPERS -----------------*/
    /** reconcatenate a split message */
    private String mashSplit(String[] split)
    {
        String toSend = split[0];
        for(int i = 1; i < split.length; i++)
            toSend = toSend + "," + split[i];
        
        return toSend;
    }
    
    /** Send a message down the tree */
    private void sendDown(String message)
    {
        for(Integer i : childrenId)
            commChannels.get(i).offer(message);
    }
    
    /** Send the message up the tree */
    private void sendUp(String message)
    {
        // root, no parent
        if(parent == null)
            return;
        // send up and store in case parent has failed
        commChannels.get(parent.nodeId).offer(message);
        savedMessages.add(message);
    }
    
}
